home *** CD-ROM | disk | FTP | other *** search
/ Personal Computer World 2008 February / PCWFEB08.iso / Software / Freeware / Miro 1.0 / Miro_Installer.exe / xulrunner / python / test / httpdownloadertest.py < prev    next >
Encoding:
Python Source  |  2007-11-12  |  6.1 KB  |  152 lines
  1. import os
  2. import tempfile
  3.  
  4. import util # This adds logging.timing
  5. import download_utils
  6. import httpclient
  7. from test import schedulertest
  8. from dl_daemon import download
  9.  
  10. def testingNextFreeFilename(filename):
  11.     return tempfile.mktemp()
  12.  
  13. class TestingDownloader(download.HTTPDownloader):
  14.     UPDATE_CLIENT_INTERVAL = 0 # every data block does an updateClient
  15.  
  16.     def __init__(self, test, *args, **kwargs):
  17.         self.test = test
  18.         if 'restore' in kwargs:
  19.             kwargs['restore']['statusCallback'] = lambda: 0
  20.             kwargs['restore']['lastStatus'] = None
  21.         else:
  22.             self.statusCallback = lambda: 0
  23.             self.lastStatus = None
  24.         download.HTTPDownloader.__init__(self, *args, **kwargs)
  25.  
  26.     def updateClient(self):
  27.         self.lastStatus = self.getStatus()
  28.         self.test.addIdle(self.statusCallback, "status callback")
  29.  
  30. class HTTPDownloaderTest(schedulertest.EventLoopTest):
  31.     def setUp(self):
  32.         super(HTTPDownloaderTest, self).setUp()
  33.         download.chatter = False
  34.         download.nextFreeFilename = testingNextFreeFilename
  35.         download._downloads = {}
  36.         allConnections = []
  37.         for conns in httpclient.HTTPClient.connectionPool.connections.values():
  38.             allConnections.extend(conns['active'])
  39.             allConnections.extend(conns['free'])
  40.         for c in allConnections:
  41.             c.closeConnection()
  42.         httpclient.HTTPClient.connectionPool = httpclient.HTTPConnectionPool()
  43.  
  44.     def tearDown(self):
  45.         download.nextFreeFilename = download_utils.nextFreeFilename
  46.         download.chatter = True
  47.         super(HTTPDownloaderTest, self).tearDown()
  48.  
  49.     def stopOnFinished(self):
  50.         if self.downloader.state == "finished":
  51.             self.stopEventLoop(False)
  52.  
  53.     def getDownloadedData(self):
  54.         return open(self.downloader.filename, 'rb').read()
  55.  
  56.     def countConnections(self):
  57.         total = 0
  58.         pool = httpclient.HTTPClient.connectionPool
  59.         for conns in pool.connections.values():
  60.             total += len(conns['active'])
  61.         return total
  62.  
  63. #    Really slow test that downloads a very large file.
  64. #    def testHuge(self):
  65. #        url = 'http://archive-c01.libsyn.com/aXdueJh2m32XeGh6l3efp5qtZXiX/podcasts/askaninja/AANQ21.m4v'
  66. #        self.downloader = TestingDownloader(url, "ID1")
  67. #        self.downloader.statusCallback = self.stopOnFinished
  68. #        self.runEventLoop(timeout=120)
  69. #        self.assertEquals(self.failed, None)
  70. #
  71.     def testDownload(self):
  72.         url = u'http://participatoryculture.org/democracytest/normalpage.txt'
  73.         self.downloader = TestingDownloader(self, url, "ID1")
  74.         self.downloader.statusCallback = self.stopOnFinished
  75.         self.runEventLoop()
  76.         self.assertEquals(self.getDownloadedData(), "I AM A NORMAL PAGE\n")
  77.  
  78.     def testStop(self):
  79.         # nice large download so that we have time to interrupt it
  80.         url = u'http://www.getdemocracy.com/images/linux-screen.jpg'
  81.         self.downloader = TestingDownloader(self, url, "ID1")
  82.         def stopOnData():
  83.             if (self.downloader.state == 'downloading' and 
  84.                     self.downloader.currentSize > 0):
  85.                 self.downloader.stop(False)
  86.                 self.stopEventLoop(False)
  87.         self.downloader.statusCallback = stopOnData
  88.         self.runEventLoop()
  89.         self.assertEquals(self.downloader.state, 'stopped')
  90.         self.assertEquals(self.downloader.currentSize, 0)
  91.         self.assert_(not os.path.exists(self.downloader.filename))
  92.         self.assertEquals(self.countConnections(), 0)
  93.         def restart():
  94.             self.downloader.start()
  95.         self.addTimeout(0.5, restart, 'restarter')
  96.         self.downloader.statusCallback = self.stopOnFinished
  97.         self.runEventLoop()
  98.         self.assertEquals(self.downloader.currentSize, 45572)
  99.         self.assertEquals(self.downloader.totalSize, 45572)
  100.  
  101.     def testPause(self):
  102.         url = u'http://www.getdemocracy.com/images/linux-screen.jpg'
  103.         self.downloader = TestingDownloader(self, url, "ID1")
  104.         def pauseOnData():
  105.             if (self.downloader.state == 'downloading' and 
  106.                     self.downloader.currentSize > 0):
  107.                 self.downloader.pause()
  108.                 self.stopEventLoop(False)
  109.         self.downloader.statusCallback = pauseOnData
  110.         self.runEventLoop()
  111.         self.assertEquals(self.downloader.state, 'paused')
  112.         self.assert_(self.downloader.currentSize > 0)
  113.         self.assert_(os.path.exists(self.downloader.filename))
  114.         self.assertEquals(self.countConnections(), 0)
  115.         def restart():
  116.             self.downloader.start()
  117.         self.addTimeout(0.5, restart, 'restarter')
  118.         self.downloader.statusCallback = self.stopOnFinished
  119.         self.runEventLoop()
  120.         self.assertEquals(self.downloader.currentSize, 45572)
  121.         self.assertEquals(self.downloader.totalSize, 45572)
  122.  
  123.     def testRestore(self):
  124.         url = u'http://www.getdemocracy.com/images/linux-screen.jpg'
  125.         self.downloader = TestingDownloader(self, url, "ID1")
  126.         def pauseInMiddle():
  127.             if (self.downloader.state == 'downloading' and 
  128.                     self.downloader.currentSize > 1000):
  129.                 self.downloader.pause()
  130.                 self.stopEventLoop(False)
  131.         self.downloader.statusCallback = pauseInMiddle
  132.         self.runEventLoop()
  133.         self.assertEquals(self.downloader.state, 'paused')
  134.         self.assert_(0 < self.downloader.currentSize < 2000)
  135.         restore = self.downloader.lastStatus.copy()
  136.         restore['state'] = 'downloading'
  137.         download._downloads = {}
  138.         self.downloader2 = TestingDownloader(self, restore=restore)
  139.         restoreSize = restore['currentSize']
  140.         self.restarted = False
  141.         def statusCallback():
  142.             # make sure we don't ever restart it
  143.             if self.downloader2.currentSize < restoreSize:
  144.                 print "%d < %d" % (self.downloader2.currentSize, restoreSize)
  145.                 self.restarted = True
  146.                 self.stopEventLoop(False)
  147.             elif self.downloader2.state == 'finished':
  148.                 self.stopEventLoop(False)
  149.         self.downloader2.statusCallback = statusCallback
  150.         self.runEventLoop()
  151.         self.assert_(not self.restarted)
  152.